package com.rpc.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Description 连接管理器
 * @Author 顾志杰
 * @Date 2022/2/28 09:23
 */
@Slf4j
public class RpcConnectManager {

  private static volatile RpcConnectManager RPC_CONNECT_MANAGER = new RpcConnectManager();

  private RpcConnectManager() {

  }

  /**
   * 单例模式
   */
  public static RpcConnectManager getInstance() {
    return RPC_CONNECT_MANAGER;
  }

  /*一个连接地址，对应一个实际的业务处理器啊(client)*/
  private Map<InetSocketAddress, RpcClientHandler> connectedHandlerMap = new HashMap<InetSocketAddress, RpcClientHandler>();

  /*用于异步的提交连接请求的线程池*/
  private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));

  /*所有连接成功的地址所对应的任务执行器列表clientHandlerList*/
  private CopyOnWriteArrayList<RpcClientHandler> connectedHandlerList = new CopyOnWriteArrayList<RpcClientHandler>();


  private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);

  private ReentrantLock connectedLock = new ReentrantLock();

  private Condition connectedCondition = connectedLock.newCondition();

  private long connectTimeoutMills = 6000;

  private volatile boolean isRunning = true;

  private volatile AtomicInteger handlerIdx = new AtomicInteger(0);
  /*1. 一步连接 线程池 真正的发起连接，连接失败监听，连接成功监听*/

  /*2. 对于连接进来的资源做一个缓存（做一个管理）updateConnectedServer*/

  /**
   * 发起连接方法
   *
   * @param serverAddress 多个连接地址用','拼接
   */
  public void connect(final String serverAddress) {
    List<String> allServerAddress = Arrays.asList(serverAddress.split(","));
    updateConnectedServer(allServerAddress);

  }

  /**
   * 更新缓存信息并异步发起连接
   *
   * @param allServerAddress 所有的连接地址
   */
  private void updateConnectedServer(List<String> allServerAddress) {
    if (CollectionUtils.isEmpty(allServerAddress)) {
      log.error("no available server address!");
      /*添加告警 清空所有信息*/
      clearConnected();
    } else {
      /*解析存储连接地址的url与port 临时存储到newAllServerNodeSet*/
      HashSet<InetSocketAddress> newAllServerNodeSet = new HashSet<InetSocketAddress>();
      allServerAddress.forEach(url -> {
        String[] array = url.split(":");
        if (array.length == 2) {
          String host = array[0];
          int port = Integer.parseInt(array[1]);
          final InetSocketAddress remotePeer = new InetSocketAddress(host, port);
          newAllServerNodeSet.add(remotePeer);
        }
      });

      /*调用连接方法 发起远程连接操作*/
      newAllServerNodeSet.forEach(serverNodeAddress -> {
        /*如果不存在才发起异步连接*/
        if (!connectedHandlerMap.containsKey(serverNodeAddress)) {
          connectAsync(serverNodeAddress);
        }
      });
      /*如果allServerAddress列表里不存在的连接地址，那么我们需要从缓存中移除*/
      connectedHandlerList.forEach(connectedCondition -> {
        SocketAddress remotePeer = connectedCondition.getRemotePeer();
        if (!newAllServerNodeSet.contains(remotePeer)) {
          log.info("remove invalid server node : {}", remotePeer);
          RpcClientHandler handler = connectedHandlerMap.get(remotePeer);
          if (handler != null) {
            handler.close();
            connectedHandlerMap.remove(remotePeer);
          }
          connectedHandlerList.remove(remotePeer);
        }
      });
    }
  }


  /**
   * 异步发起连接的方法
   */
  private void connectAsync(InetSocketAddress remotePeer) {
    /*new Runnable()*/
    threadPoolExecutor.submit(() -> {
      Bootstrap b = new Bootstrap();
      b.group(eventLoopGroup)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .handler(new RpcClientInitializer());
      connect(b, remotePeer);
    });

  }

  /**
   * 连接实现
   */
  private void connect(final Bootstrap b, InetSocketAddress remotePeer) {
    /*建立连接*/
    final ChannelFuture channelFuture = b.connect(remotePeer);

    /*连接失败的时候添加监听 清除资源后进行发起重连操作 new ChannelFutureListener()*/
    channelFuture.channel().closeFuture().addListener((ChannelFutureListener) future -> {
      log.info("channelFuture.channel close operationComplete,remote per = {}", remotePeer);
      /*new Runnable()*/
      future.channel().eventLoop().schedule(() -> {
        log.warn("connect fail, to reconnect!");
        clearConnected();
        connect(b, remotePeer);
      }, 3, TimeUnit.SECONDS);
    });

    /*连接成功的时候添加监听 把我们的新连接放入缓存中 new ChannelFutureListener()*/
    channelFuture.addListener((ChannelFutureListener) future -> {
      if (future.isSuccess()) {
        log.info("successfully connect to remote server,remote peer = {}", remotePeer);
        RpcClientHandler handler = future.channel().pipeline().get(RpcClientHandler.class);
        addHandler(handler);
      }
    });
  }

  /**
   * 添加RpcClientHandler到指定的缓存中
   * connectedHandlerList & connectedHandlerMap
   */
  private void addHandler(RpcClientHandler handler) {
    connectedHandlerList.add(handler);
    InetSocketAddress remotePeer = (InetSocketAddress) handler.getRemotePeer();
    connectedHandlerMap.put(remotePeer, handler);
    /*signalAvailableHandler 唤醒可用的业务执行器*/
    signalAvailableHandler();
  }

  /**
   * 连接失败的时候 及时的释放资源 清空缓存
   * 先删除所有的connectedHandlerMap中的数据
   * 再删除connectedHandlerList中的数据
   */
  private void clearConnected() {
    for (final RpcClientHandler rpcClientHandler : connectedHandlerList) {
      /*通过RpcClientHandler 找到具体的remotePeer，从connectedHandlerMap进行移除指定的RpcClientHandler*/
      SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
      RpcClientHandler handler = connectedHandlerMap.get(remotePeer);
      if (handler != null) {
        handler.close();
        connectedHandlerMap.remove(remotePeer);
      }
    }
    connectedHandlerList.clear();
  }

  /**
   * 唤醒另外一端的线程(阻塞的状态中) 告知有新连接接入
   */
  private void signalAvailableHandler() {
    connectedLock.lock();
    try {
      connectedCondition.signalAll();
    } finally {
      connectedLock.unlock();
    }
  }

  /**
   * 等待新连接接入通知方法
   *
   * @return
   * @throws InterruptedException
   */
  private boolean waitingForAvailableHandler() throws InterruptedException {
    connectedLock.lock();
    try {
      return connectedCondition.await(this.connectTimeoutMills, TimeUnit.MICROSECONDS);
    } finally {
      connectedLock.unlock();
    }
  }


  /**
   * 选择一个实际的业务处理器
   *
   * @return
   */
  public RpcClientHandler chooseHandler() {
    CopyOnWriteArrayList<RpcClientHandler> handlers = (CopyOnWriteArrayList<RpcClientHandler>) this.connectedHandlerList.clone();
    int size = handlers.size();
    while (isRunning && size <= 0) {
      try {
        boolean available = waitingForAvailableHandler();
        if (available) {
          handlers = (CopyOnWriteArrayList<RpcClientHandler>) this.connectedHandlerList.clone();
          size = handlers.size();
        }
      } catch (InterruptedException e) {
        log.error(" wating for available node is interrupted !");
        throw new RuntimeException("no connect any servers!", e);
      }
    }
    if (!isRunning) {
      return null;
    }
    /*最终使用取模方式取得其中一个业务处理器进行实际的业务处理*/
    return handlers.get((handlerIdx.getAndAdd(1) + size) % size);
  }

  /**
   * 停止
   */
  public void stop() {
    isRunning = false;
    connectedHandlerList.forEach(connectedCondition -> {
      connectedCondition.close();
    });
    /*在这里要调用一下唤醒操作*/
    signalAvailableHandler();
    threadPoolExecutor.shutdown();
    eventLoopGroup.shutdownGracefully();
  }

  /**
   * 发起重连的方法  需要把对应的资源释放
   */
  public void reconnect(final RpcClientHandler handler, SocketAddress remotePeer) {
    if (handler != null) {
      handler.close();
      connectedHandlerList.remove(handler);
      connectedHandlerMap.remove(remotePeer);
    }
    connectAsync((InetSocketAddress) remotePeer);
  }

}
